﻿using Leo.Common.MQ;
using Microsoft.Extensions.DependencyInjection;

namespace Leo.Common.Kafka
{
    public class KafkaBuilder
    { 
        internal KafkaProducer Producer;

        internal KafkaSetting Setting;

        private readonly IServiceCollection services;

        internal readonly List<MQConsumerBase> Consumers = new List<MQConsumerBase>();

        internal readonly List<KafkaConsumerWrapper> wrappers = new List<KafkaConsumerWrapper>();

        public KafkaBuilder(KafkaSetting setting, IServiceCollection services)
        {
            this.Setting = setting;
            this.services = services;
        }
 
        public KafkaBuilder AddConsumer<T>(KafkaConsumerConfig config) where T : IMessageHandler
        {
            KafkaConsumerWrapper kafkaConsumerWrapper = this.wrappers.FirstOrDefault((KafkaConsumerWrapper x) => x.Config == config);
            if (kafkaConsumerWrapper == null)
            {
                kafkaConsumerWrapper = new KafkaConsumerWrapper(config);
                this.wrappers.Add(kafkaConsumerWrapper);
            }
            Type typeOfT = typeof(T);
            if (!kafkaConsumerWrapper.HandlerTypes.Any((Type x) => x == typeOfT))
            {
                kafkaConsumerWrapper.HandlerTypes.Add(typeOfT);
            }
            return this;
        }
         
        public KafkaBuilder AddConsumer<T>(string topic, string groupId, string clientId = null) where T : IMessageHandler
        {
            KafkaConsumerConfig config = new KafkaConsumerConfig
            {
                Topic = topic,
                GroupId = groupId,
                ClientId = clientId
            };
            return this.AddConsumer<T>(config);
        }         
        
        public KafkaBuilder AddProducer()
        {
            if (this.Producer == null)
            {
                this.Producer = new KafkaProducer(this.Setting);
            }
            return this;
        }
       
        public IMessageQueueProvider Build()
        {
            KafkaProvider kafkaProvider = new KafkaProvider(this);
            ServiceCollectionServiceExtensions.AddSingleton<KafkaBuilder>(this.services, this);
            ServiceCollectionServiceExtensions.AddSingleton<IMessageQueueProvider>(this.services, kafkaProvider);
            ServiceCollectionServiceExtensions.AddSingleton<KafkaMessageQueueProviderBase>(this.services, kafkaProvider);
            if (this.Producer != null)
            {
                ServiceCollectionServiceExtensions.AddSingleton<IMQProducer>(this.services, this.Producer);
                ServiceCollectionServiceExtensions.AddSingleton<IKafkaProducer>(this.services, this.Producer);
            }
            ServiceCollectionHostedServiceExtensions.AddHostedService<KafkaConsumerHostedService>(this.services);
            foreach (Type type in Enumerable.ToList<Type>(this.wrappers.SelectMany((KafkaConsumerWrapper x) => x.HandlerTypes).Distinct<Type>()))
            {
                ServiceCollectionServiceExtensions.AddSingleton(this.services, type);
            }
            return kafkaProvider;
        }
       
      
    }
}
